package fregat.base.database;

import oracle.jdbc.driver.OracleConnection;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.sql.*;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.ResourceBundle;

import fregat.base.util.ConfigurationBundle;
import fregat.base.util.Logger;
import fregat.base.util.StopWatch;

/**
 * Oracle specific wrapper for a JDBC database connection to an Oracle database.
 */
public class DatabaseConnector {
  private Connection connection;
  private Hashtable statements;
  private String hostname;
  private String user;
  private String password;
  private int port;
  private Sid sid;
  private ServiceName serviceName;
  private Logger logger;
  private CallableStatement endTransactionStatement;
  private boolean userPrefix;

  private long reconnect = 5000;
  private boolean shutdown;
  private static String ALERT_PREFIX = "ALERT_";
  private static ResourceBundle resource = ResourceBundle.getBundle("fregat.base.database.resource");

  /**
   * @return Instance of database connection with exactly the same parameters.
   */
  public DatabaseConnector createInstance() {
    return new DatabaseConnector(hostname, user, password, sid, port, logger);
  }

  /**
   * This string is returned if the alert timeout was reached in
   * <code>waitForAlert</code>.
   */
  public static final String ALERT_TIMEOUT = "TIMEOUT";

  /**
   * Creates a new database connector for the specified parameters.
   *
   * @param hostname Name of database server
   * @param user     Name of database user
   * @param password Password of database user
   * @param sid      Service identifier of database server
   * @param port     Service port of database server
   */
  public DatabaseConnector(String hostname, String user,
                           String password, Sid sid, int port) {
    this(hostname, user, password, sid, port, Logger.getDefaultCategory());
  }

  /**
   * Creates a new database connector for the specified parameters.
   *
   * @param hostname Name of database server
   * @param user     Name of database user
   * @param password Password of database user
   * @param sid      Service identifier of database server
   * @param port     Service port of database server
   * @param logger   Logging instance
   */
  public DatabaseConnector(String hostname, String user,
                           String password, Sid sid, int port, Logger logger) {
    this.hostname = hostname;
    this.user = user;
    this.password = password;
    this.sid = sid;
    this.port = port;
    this.logger = logger;
    serviceName = new ServiceName(null);
  }
  
  public DatabaseConnector(String hostname, String user,
          String password, ServiceName servicename, int port, Logger logger) {
this.hostname = hostname;
this.user = user;
this.password = password;
this.serviceName = servicename;
this.port = port;
this.logger = logger;
sid = new Sid(null);

}

  /**
   * Creates a new database connector for the specified parameters.
   *
   * @param connectionString Oracle connection string
   */
  public DatabaseConnector(String connectionString) {
    parseConnectionString(connectionString);
  }

  /**
   * Creates a new database connector for the specified parameters.
   *
   * @param hostname  Name of database server
   * @param user      Name of database user
   * @param password  Password of database user
   * @param sid       Service identifier of database server
   * @param port      Service port of database server
   * @param reconnect Timeout before reconnection
   */
  public DatabaseConnector(String hostname, String user,
                           String password, Sid sid, int port, long reconnect) {
    this(hostname, user, password, sid, port, Logger.getDefaultCategory());
    this.reconnect = reconnect;
  }


  /**
   * Creates a new database connector for the specified parameters.
   *
   * @param connectionString Oracle connection string
   * @param reconnect        Timeout before reconnection
   */
  public DatabaseConnector(String connectionString, long reconnect) {
    this(connectionString);
    this.reconnect = reconnect;
  }

  /**
   * Creates a new database connector by reading the configuration from
   * the specified configuration bundle using the following keys:<br>
   * &lt;configPath&gt;.db.hostname for database hostname (default: lms)
   * &lt;configPath&gt;.db.user for schema user (default: sysadm)
   * &lt;configPath&gt;.db.password for schema password (default: schema user)
   * &lt;configPath&gt;.db.reconnect for reconnect timeout (default: 5000 ms)
   *
   * @param configuration Configuration bundle
   * @param configPath    Path to database configuration
   */
  public DatabaseConnector(ConfigurationBundle configuration, String configPath) {
    this(configuration, configPath, Logger.getDefaultCategory());
  }

  /**
   * Creates a new database connector by reading the configuration from
   * the specified configuration bundle using the following keys:<br>
   * &lt;configPath&gt;.db.hostname for database hostname (default: lms)
   * &lt;configPath&gt;.db.user for schema user (default: sysadm)
   * &lt;configPath&gt;.db.password for schema password (default: schema user)
   * &lt;configPath&gt;.db.reconnect for reconnect timeout (default: 5000 ms)
   *
   * @param configuration Configuration bundle
   * @param configPath    Path to database configuration
   * @param logger        Logging instance
   */
  public DatabaseConnector(ConfigurationBundle configuration, String configPath, Logger logger) {
    if (configPath == null) {
      configPath = "";
    }

    if (!configPath.equals("")) {
      configPath += ".";
    }

    hostname = configuration.getConfigurationString(configPath + "db.name");
    user = configuration.getConfigurationString(configPath + "db.user");
    password = configuration.getConfigurationString(configPath + "db.password");

    String connectionString = configuration.getConfigurationString(configPath + "db.connectionString");

    if (connectionString != null) {
      parseConnectionString(connectionString);
    } else {

      port = configuration.getConfigurationInteger(configPath + "db.port");
      sid.reset(configuration.getConfigurationString(configPath + "db.sid"));
      if (sid.isNull()) {
        serviceName.reset(configuration.getConfigurationString(configPath + "db.service"));
      }
    }

    long reconnect = configuration.getConfigurationLong(configPath + "db.reconnect");

    // set defaults
    if (hostname == null) {
      hostname = "lms";
    }
    if (user == null) {
      user = "sysadm";
    }
    if (password == null) {
      password = user;
    }
    if (port < 0) {
      port = 1521;
    }
    if (sid.isNull() && serviceName.isNull()) {
      sid.reset("lms");
    }
    if (reconnect > 0) {
      this.reconnect = reconnect;
    }

    this.logger = logger;
  }

  private String parseValue(String key, String connectionString) {
    int start = connectionString.indexOf(key);
    if (start > 0) {
      start = connectionString.indexOf("=", start);
      int end = connectionString.indexOf(")", start);
      return connectionString.substring(start + 1, end).trim();
    }
    return null;
  }

  private void parseConnectionString(String connectionString) {
    // supported connection strings
    //(DESCRIPTION=(ADDRESS=(PROTOCOL = TCP)(HOST = <hostname> )(PORT = <port>))(CONNECT_DATA = (SID = <sid>)))
    //(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=<hostname>)(PORT=<port>)))(CONNECT_DATA =(SERVICE_NAME=<service name>)))
    hostname = parseValue("HOST", connectionString);
    sid.reset(parseValue("SID", connectionString));

    if (sid.isNull()) {
      serviceName.reset(parseValue("SERVICE_NAME", connectionString));
    }

    port = Integer.parseInt(parseValue("PORT", connectionString));
  }


  /**
   * Connects to the database and overrides the parameters with the specified ones.
   *
   * @param name     TNS hostname of database
   * @param user     Name of database user
   * @param password Password of database user
   * @return True if connectionString successful
   */
  public boolean connectToDatabase(String name, String user, String password) {
    this.hostname = name;
    this.user = user;
    this.password = password;
    return connectToDatabase();
  }

  /**
   * Connects to the database specified in constructor.
   *
   * @return True if connectionString successful
   */
  public boolean connectToDatabase() {
    return this.connectToDatabase(true);
  }

  /**
   * Connects to the database specified in constructor.
   *
   * @param reconnect True, if connection should be retried
   * @return True if connectionString successful
   */
  public boolean connectToDatabase(boolean reconnect) {
    this.connection = createConnection(reconnect, true);

    // check if event queue is enabled
    try {
      CallableStatement statement = connection.prepareCall("begin ? := lvs.getwertsysein('SWITCH','EVENTQUEUE'); end;");
      statement.registerOutParameter(1, Types.INTEGER);

      statement.execute();

      int result = statement.getInt(1);

      userPrefix = true;

      if (result == 1) {
        endTransactionStatement = prepareStatement("ENDTA", "begin sess.endTA(true); end;");
      }

    } catch (SQLException ex) {
      logger.debug(resource.getString("DatabaseConnector_createConnection_3"));

    }

    return (connection != null);
  }

  /**
   * @return True, if database exists.
   */
  public boolean databaseExists() {
    return createConnection(false, false) != null;
  }

  /**
   * Waits for the specified alert or times out after given time.
   *
   * @param alertName Name of alert
   * @param timeout   Timeout in [s]
   * @return Alert message string or ALERT_TIMEOUT if timed out
   */
  public String waitForAlert(String alertName, int timeout) {
    String key = ALERT_PREFIX + alertName;

    alertName = addUserPrefix(alertName);

    try {
      CallableStatement statement = null;
      if (!getStatements().containsKey(key)) {
        statement = prepare("begin dbms_alert.register(?); end;");
        statement.setString(1, alertName);
        execute(statement, true, false);

        statement = prepare("begin dbms_alert.waitone(?,?,?,?); end;");
        statement.setString(1, alertName);
        statement.registerOutParameter(2, Types.CHAR);
        statement.registerOutParameter(3, Types.INTEGER);
        getStatements().put(key, statement);
      } else {
        statement = (CallableStatement) getStatements().get(key);
      }

      logger.note(resource.getString("DatabaseConnector_waitForAlert_1").replaceFirst("%1", alertName));

      statement.setInt(4, timeout);
      execute(statement, true, false);

      if (statement.getInt(3) > 0) {
        return ALERT_TIMEOUT;
      } else {
        return statement.getString(2);
      }

    } catch (SQLException ex) {
      removeStatement(key);
      error("DatabaseConnector.waitForAlert", ex);
    }
    return null;
  }

  /**
   * Adds the database user as prefix before the alert or pipe name,
   * if it's not added before and userPrefix is set true
   *
   * @param itemName Name of alert or pipe
   * @return <database user>.<itemName name>
   */
  private String addUserPrefix(String itemName) {
    if (userPrefix) {
      if (itemName.toUpperCase().indexOf(".") == -1) {
        String name = user.toUpperCase() + "." + itemName;
        if (name.length() > 30) {
          logger.error(resource.getString("DatabaseConnector_addUserPrefix_1").replaceFirst("%1", itemName));
        } else {
          itemName = name;
        }
      }
    }
    logger.debug("itemName: "+ itemName);
    return itemName;
  }

  /**
   * Fires the specified alert to the database using a new connection.
   *
   * @param alertName Name of alert
   * @param message   Message of alert
   * @return True if successful, false on any exception
   */
  public boolean fireAlert(String alertName, String message) {
    Connection connection = createConnection(false, false);
    alertName = addUserPrefix(alertName);

    if (connection != null) {
      CallableStatement statement = null;
      try {
        statement = prepare("begin dbms_alert.signal(?,?); end;", connection);
        statement.setString(1, alertName);
        statement.setString(2, message);
        execute(statement, true, false);
        logger.note(resource.getString("DatabaseConnector_fireAlert_1").replaceFirst("%1", alertName).replaceFirst("%2", message));
        return true;

      } catch (SQLException ex) {
        error("DatabaseConnector.fireAlert('" + alertName + "','" + message + "'", ex);
      } finally {
        try {
          if (statement != null) {
            statement.close();
          }
          connection.close();
        } catch (Exception ex) {
          logger.error("DatabaseConnector.fireAlert");
        }
      }
    }
    return false;
  }

  /**
   * Prepares the call and stores the statement with the given <code>key</code>.
   *
   * @param key Key used in <code>getStatement</code>
   * @param sql SQL statement
   * @return Statement
   * @throws SQLException
   */
  public CallableStatement prepareStatement(String key, String sql) throws SQLException {
    CallableStatement statement = prepare(sql);
    getStatements().put(key, statement);
    return statement;
  }

  /**
   * Delivers the statement stored with the given <code>key</code>.
   *
   * @param key Key of statement
   * @return Statement, <code>null</code> if not existing
   */
  public CallableStatement getStatement(String key) {
    CallableStatement statement = (CallableStatement) getStatements().get(key);
    if (statement == null) {
      logger.warn(resource.getString("DatabaseConnector_getStatement_1").replaceFirst("%1", key));
    }
    return statement;
  }

  /**
   * Removes the statement with the given  <code>key</code>.
   *
   * @param key Key of statement
   */
  public void removeStatement(String key) {
    CallableStatement statement = (CallableStatement) getStatements().get(key);
    if (statement != null) {
      try {
        statement.close();
      } catch (SQLException ex) {
        error("DatabaseConnector.removeStatement", ex);
      }
      getStatements().remove(key);
    }
  }

  /**
   * Executes the statement with the given key.<br>
   * Use <code>prepareStatement</code> to prepare statement for the execution.
   *
   * @param key Key of statement
   * @return Time, database needed to execute.
   * @throws SQLException
   */
  public long executeStatement(String key) throws SQLException {
    StopWatch.resetTime(key);
    CallableStatement statement = getStatement(key);
    if (statement != null) {
      try {
        execute(statement, false, false);
      } catch (SQLException ex) {
        removeStatement(key);
        throw (ex);
      }
    }
    return StopWatch.takeTime(key);
  }

  /**
   * Eexcutes the query statement with the given key.<br>
   * Use <code>prepareStatement</code> to prepare statement for the execution.
   *
   * @param key Key of statement
   * @return The result set of the query (empty enumeration if no rows fetched,
   *         <code>null</code> if no statement prepared)
   * @throws SQLException
   */
  public ResultSet executeQuery(String key) throws SQLException {
    CallableStatement statement = getStatement(key);
    if (statement != null) {
      try {
        return execute(statement, false, true);

      } catch (SQLException ex) {
        removeStatement(key);
        throw (ex);
      }
    }
    return null;
  }

  public void commit() {
    try {
      if (endTransactionStatement != null) {
        endTransactionStatement.execute();
      } else {
        connection.commit();
      }

    } catch (SQLException ex) {
      error(resource.getString("DatabaseConnector_commit_1"), ex);
    }
  }

  public void rollback() {
    try {
      connection.rollback();
    } catch (SQLException ex) {
      error(resource.getString("DatabaseConnector_rollback_1"), ex);
    }
  }

  /**
   * Shutdown connector (including database connection and alerts).
   */
  public synchronized void shutdown() {
    shutdown = true;
    notify();
    unregisterAlerts();
    disconnectFromDatabase();
  }

  /**
   * @return JDBC database connection.
   */
  public Connection getConnection() {
    return connection;
  }

  /**
   * @return True, if connected to database
   */
  public boolean isConnected() {
    if (connection != null) {
      try {
        return !connection.isClosed();
      } catch (SQLException ex) {
        ex.printStackTrace();
      }
    }
    return false;
  }

  /**
   * Creates and opens a connection to the database defined in constructor.
   * The connector tries to reconnect if connection fails.
   *
   * @return Connection instance, null on any exception
   */
  public Connection createConnection() {
    return createConnection(true, true);
  }

  /**
   * Creates and opens a connection to the database defined in constructor.
   *
   * @param reconnect  True, if connector should try to reconnect if connection fails
   * @param logSuccess True, if message should be logged after successful connectionString
   * @return Connection instance, null on any exception
   */
  public synchronized Connection createConnection(boolean reconnect, boolean logSuccess) {
    String serviceString;
    if(!sid.isNull()){
      serviceString = "SID=" + sid.toString();
    }
    else{
      serviceString = "SERVICE_NAME=" + serviceName.toString();
    }
    String connect = "jdbc:oracle:thin:@" + hostname + ":" + port + ":" + "xe";
//          "(DESCRIPTION=(ADDRESS=(HOST=" + hostname + ")" +
//          "(PROTOCOL=tcp)(PORT=" + port + "))(CONNECT_DATA=(" + serviceString + ")))";
    
    Connection connection = null;

    do {
      String debug = resource.getString("DatabaseConnector_createConnection_1");
      try {
        DriverManager.registerDriver(new oracle.jdbc.driver.OracleDriver());

        connection = DriverManager.getConnection(connect, user, password);
        ((OracleConnection) connection).setDefaultRowPrefetch(100);
        connection.setAutoCommit(false);

        debug = debug.replaceFirst("%1", toString());
        if (logSuccess) {
          logger.info(debug);
        }

        return connection;

      } catch (SQLException ex) {
        debug = resource.getString("DatabaseConnector_createConnection_2");
        debug = debug.replaceFirst("%1", toString());
        logger.error(debug);
      }
      if (connection == null || !isConnected()) {
        try {
          wait(this.reconnect);
        } catch (InterruptedException ex) {
          ex.printStackTrace();
        }
      }
    } while (reconnect && connection == null && !shutdown);

    return null;
  }

  /**
   * Sets module of the database session.
   *
   * @param module     Module hostname
   * @param connection Database connection
   */
  public static void setModule(String module, Connection connection) {
    PreparedStatement statement = null;
    try {
      statement = connection.prepareStatement("begin dbms_application_info.set_module(?,?); end;");
      statement.setString(1, module);
      statement.setString(2, null);
      statement.execute();

    } catch (SQLException ex) {
      Logger.getDefaultCategory().error("DatabaseConnector.setModule", ex);
    } finally {
      try {
        if (statement != null) {
          statement.close();
        }
      } catch (Exception ex) {
        Logger.getDefaultCategory().error("DatabaseConnector.setModule(close)", ex);
      }
    }
  }

  /**
   * Sets information for database session
   * (<hostname>@<hostname> as module and <path> as client information).
   *
   * @param name       Name of application the connection is used for
   * @param connection Database connection
   */
  public static void setName(String name, Connection connection) {
    try {
      setModule(name + "@" + InetAddress.getLocalHost().getHostName(), connection);
      setClientInfo(System.getProperty("user.dir"), connection);
    } catch (UnknownHostException ex) {
      Logger.getDefaultCategory().error("DatabaseConnector.setName", ex);
    }
  }

  /**
   * Sets information of the database session(<hostname>@<hostname>).
   *
   * @param name Name of application the connection is used for
   */
  public void setName(String name) {
    setName(name, connection);
  }

  /**
   * Sets action of the database session.
   *
   * @param action     Name of action
   * @param connection Database connection
   */
  public static void setAction(String action, Connection connection) {
    PreparedStatement statement = null;
    try {
      statement = connection.prepareStatement("begin dbms_application_info.set_action(?); end;");
      statement.setString(1, action);
      statement.execute();

    } catch (Exception ex) {
      Logger.getDefaultCategory().error("DatabaseConnector.setAction", ex);

    } finally {
      try {
        statement.close();
      } catch (Exception ex) {
        Logger.getDefaultCategory().error("DatabaseConnector.setAction(close)", ex);
      }
    }
  }

  /**
   * Sets the client information of the database session.
   *
   * @param info       Client information
   * @param connection Database connection
   */
  public static void setClientInfo(String info, Connection connection) {
    PreparedStatement statement = null;
    try {
      statement = connection.prepareStatement("begin dbms_application_info.set_client_info(?); end;");
      statement.setString(1, info);
      statement.execute();

    } catch (Exception ex) {
      Logger.getDefaultCategory().error("DatabaseConnector.setClientInfo", ex);

    } finally {
      try {
        statement.close();
      } catch (Exception ex) {
        Logger.getDefaultCategory().error("DatabaseConnector.setClientInfo(close)", ex);
      }
    }
  }

  /**
   * Sets the client information of the database session.
   *
   * @param info Client information
   */
  public void setClientInfo(String info) {
    setClientInfo(info, connection);
  }

  /**
   * Sets module of the database session.
   *
   * @param module Name of Module
   */
  public void setModule(String module) {
    setModule(module, connection);
  }

  /**
   * Sets action of the database session.
   *
   * @param action Name of action
   */
  public void setAction(String action) {
    setAction(action, connection);
  }


  /**
   * @return Table with stored statements
   */
  private Hashtable getStatements() {
    if (statements == null) statements = new Hashtable();
    return statements;
  }

  /**
   * Closes the database connection.
   */
  private void disconnectFromDatabase() {
    try {
      if (isConnected()) {
        connection.close();
      }
    } catch (SQLException ex) {
      error("DatabaseConnector.disconnectFromDatabase", ex);
    } finally {
      statements.clear();
      String debug = resource.getString("DatabaseConnector_disconnectFromDatabase_1");
      debug = debug.replaceFirst("%1", toString());
      logger.info(debug);
    }
  }

  /**
   * Disconnects the connector from all database alerts.
   */
  private void unregisterAlerts() {
    Enumeration e = getStatements().keys();
    while (e.hasMoreElements()) {
      String key = (String) e.nextElement();
      if (key.startsWith(ALERT_PREFIX)) {
        String alertName = key.substring(ALERT_PREFIX.length(), key.length());
        try {
          CallableStatement statement = prepare("begin dbms_alert.remove(?); end;");
          statement.setString(1, alertName);
          execute(statement, true, false);

        } catch (SQLException ex) {
          error("DatabaseConnector.unregisterAlerts", ex);
        }
      }
    }
  }

  private ResultSet execute(CallableStatement statement, boolean update, boolean query) throws SQLException {
    try {
      if (query) {
        return statement.executeQuery();
      }

      if (update) {
        statement.executeUpdate();
      } else {
        statement.execute();
      }
      return null;

    } catch (SQLException ex) {
      handleException(ex);
      throw (ex);
    }
  }

  private CallableStatement prepare(String sql) throws SQLException {
    CallableStatement statement = null;
    do {
      statement = prepare(sql, connection);
    } while (statement == null && !shutdown);
    return statement;
  }

  private CallableStatement prepare(String sql, Connection connection) throws SQLException {
    try {
      return connection.prepareCall(sql);
    } catch (SQLException ex) {
      if (!handleException(ex)) {
        throw (ex);
      }
    }
    return null;
  }

  private boolean handleException(SQLException ex) {
    switch (ex.getErrorCode()) {
      case 17002:
      case 17410:
      case 1089: // immediate shutdown in progress
      case 1090: // shutdown in progress
      case 1092: // shutdown abort in progress
      case 6508: // couldn't find program unit being called
      case 468:  // existing state of packages has been discarded  
        disconnectFromDatabase();
        connectToDatabase();
        return true;
      default:
        return false;
    }
  }

  private void error(String text, SQLException ex) {
    logger.error(text + " (ORA-" + ex.getErrorCode() + ")", ex);
  }

  /**
   * Replaces the parameters in the SQL statement with the values from the
   * array.
   *
   * @param sql   SQL statement with '?' as parameter placeholder
   * @param param Array with parameter values
   * @return SQL text with all parameter replaced
   */
  public static String logStatement(String sql, String[] param) {
    String log = sql.replace('?', '%');
    for (int i = 0; i < param.length; i++) {
      log = log.replaceFirst("%", param[i].replaceAll("\\$", "_"));
    }
    return log;
  }

  /**
   * String representation of connection
   *
   * @return <user>@<database>
   */
  public String toString() {
    return user + "@" + hostname;
  }

  /**
   * Writes the given message to the specified pipe.
   *
   * @param pipeName Name of pipe
   * @param message  Message to be written
   */
  public void writePipe(String pipeName, String message) {
    Connection connection = createConnection();

    pipeName = addUserPrefix(pipeName);

    StringBuffer sql = new StringBuffer();
    sql.append("begin dbms_pipe.pack_message('");
    sql.append(message);
    sql.append("'); ? := dbms_pipe.send_message('");
    sql.append(pipeName);
    sql.append("',10); end;");

    try {
      CallableStatement statement = connection.prepareCall(sql.toString());
      statement.registerOutParameter(1, Types.NUMERIC);
      statement.execute();
      connection.close();

    } catch (SQLException ex) {
      logger.error("DatabaseConnector.writePipe", ex);
    }
  }

  /**
   * @param pipeName Name of pipe
   * @param timeout  Read timeout in [ms]
   * @return Message read from pipe
   */
  public String readPipe(String pipeName, int timeout) {
    String message = null;

    pipeName = addUserPrefix(pipeName);

    try {
      CallableStatement statement = getStatement("RECEIVE");
      if (statement == null) {
        statement = prepareStatement("RECEIVE", "begin ? := dbms_pipe.receive_message(?,?); end;");

        statement.registerOutParameter(1, Types.INTEGER);
        statement.setString(2, pipeName);
        statement.setInt(3, timeout);
      }
      executeStatement("RECEIVE");
      int error = statement.getInt(1);

      if (error == 0) {
        statement = getStatement("UNPACK");
        if (statement == null) {
          statement = prepareStatement("UNPACK", "begin dbms_pipe.unpack_message(?); end;");
          statement.registerOutParameter(1, Types.VARCHAR);
        }
        executeStatement("UNPACK");
        message = statement.getString(1);
      }
    } catch (SQLException ex) {
      logger.error("DatabaseConnector.readPipe", ex);
    }
    return message;
  }
}
